package com.zhuoli.service.springboot.kafka.producer.controller;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.zhuoli.service.springboot.kafka.producer.common.request.SendKafkaRequest;
import com.zhuoli.service.springboot.kafka.producer.service.MessageControllerService;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;

/**
 * @Author: zhuoli
 * @Date: 2018/9/4 10:52
 * @Description:
 */
@RestController
@AllArgsConstructor
@RequestMapping("/kafka")
@Slf4j
public class MessageController {

    private MessageControllerService messageControllerService;

    private ObjectMapper objectMapper;

    @RequestMapping(value = "/send", method = RequestMethod.POST)
    public ResponseEntity sendKafka(@RequestBody SendKafkaRequest request) {
        try {
            request.setSendTime(new Date());
            log.info("kafka的消息={}", objectMapper.writeValueAsString(request));
            messageControllerService.sendMessage("zhuoli", objectMapper.writeValueAsString(request));
            log.info("发送kafka成功.");
            return ResponseEntity.status(HttpStatus.OK).body("message send success");
        } catch (Exception e) {
            log.info("Exception is {}", e);
            return ResponseEntity.status(HttpStatus.EXPECTATION_FAILED).body("exception");
        }
    }
}
